package net.wicp.tams.common.binlog.dump.listener;

import java.util.Map;
import java.util.Properties;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.bulk.BulkItemResponse;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.LoggerUtil;
import net.wicp.tams.common.binlog.dump.bean.DumpEvent;
import net.wicp.tams.common.constant.JvmStatus;
import net.wicp.tams.common.es.Action;
import net.wicp.tams.common.es.EsData;
import net.wicp.tams.common.es.EsData.Builder;
import net.wicp.tams.common.es.EsObj;
import net.wicp.tams.common.es.RelaValue;
import net.wicp.tams.common.es.UpdateSet;
import net.wicp.tams.common.es.bean.MappingBean;
import net.wicp.tams.common.es.bean.MappingBean.Propertie;
import net.wicp.tams.common.es.client.ESClient;
import net.wicp.tams.common.es.client.singleton.ESClientOnlyOne;
import net.wicp.tams.common.es.client.threadlocal.EsClientThreadlocal;

/****
 * 需要创建好索引eg: IndexParamsBean indexParamsBean =
 * IndexParamsBean.builder().db("test").tb("user_info").db1("test")
 * .tb1("user_addr").rela1("user_id").build();
 * System.out.println(indexParamsBean); Result result =
 * ESClientOnlyOne.getInst().getESClient().indexCreate(conn, indexParamsBean,
 * "test2", 2, 0);
 * 
 * @author andy.zhou
 *
 */
@Slf4j
public class SenderEs implements IBusiSender<DumpEvent> {
	private static final org.slf4j.Logger errorlog = org.slf4j.LoggerFactory.getLogger("errorBinlog");
	private Properties props;
	private JSONObject relaObj;// relaObj.isEmpty 是已初始化但没有关联关系的索引
	private static volatile boolean isInit = false;

	@Override
	public void init(String dumpId) {
		String configOriPre = "common.binlog.alone.dump.destination." + dumpId + ".es";
		this.props = Conf.getPreToProp(configOriPre, true);
		if (!isInit) {
			synchronized (this) {
				if (!isInit) {
					// 覆盖属性值
					Properties overProps = Conf.replacePre("common.binlog.alone.dump.global.es", "common.es");
					Conf.overProp(overProps);
					isInit = true;
				}
			}
		}

		// 初始化关联关系
		if (relaObj == null) {
			synchronized (SenderEs.class) {
				if (relaObj == null) {
					Map<String, Propertie> queryMapping_tc_all = ESClientOnlyOne.getInst().getESClient()
							.queryMapping_tc_all(this.props.getProperty("index.name"),
									this.props.getProperty("index.type"));
					if (queryMapping_tc_all.containsKey(Conf.get("common.es.assit.rela.key"))) {
						relaObj = queryMapping_tc_all.get(Conf.get("common.es.assit.rela.key")).getRelations();
					} else {
						relaObj = new JSONObject();
					}
				}
			}
		}
	}

	@Override
	public void doSend(DumpEvent dataBuilders) {
		ESClient eSClient = EsClientThreadlocal.createPerThreadEsClient();
		Builder esDataBuilder = EsData.newBuilder();
		esDataBuilder.setIndex(props.getProperty("index.name"));
		esDataBuilder.setType(props.getProperty("index.type"));
		esDataBuilder.setAction(Action.update);
		esDataBuilder.setUpdateSet(UpdateSet.newBuilder().setUpsert(true).build());
		String[] primarys = dataBuilders.getDump().getPrimarys();
		boolean hasprimarys = ArrayUtils.isNotEmpty(primarys);
		for (Map<String, String> datamap : dataBuilders.getDatas()) {
			EsObj.Builder esObjBuilder = EsObj.newBuilder();
			esObjBuilder.putAllSource(datamap);
			if (hasprimarys) {
				String[] values = new String[primarys.length];
				for (int j = 0; j < values.length; j++) {
					values[j] = datamap.get(primarys[j]);
				}
				String idstr = CollectionUtil.arrayJoin(values, "-");
				if (StringUtils.isEmpty(idstr)) {
					log.error("id是空值");
					continue;
				}
				// 关联关系支持 20181201
				String tablename = dataBuilders.getDump().getTb();
				boolean isroot = MappingBean.isRoot(relaObj, tablename);
				if (isroot) {// 根元素或是没有关联关联的索引
					esObjBuilder.setId(idstr);
					if (relaObj != null && !relaObj.isEmpty()) {
						esObjBuilder.setRelaValue(RelaValue.newBuilder().setName(tablename));// tams_relations
					}
				} else {// 有关联关系且不是根元素
					String relaName = MappingBean.getRelaName(relaObj, tablename);
					String[] relaNameAry = relaName.split(":");
					String parentId = datamap.get(relaNameAry[1]);
					esObjBuilder.setId(String.format("%s:%s", tablename, idstr));// 有可能与主表id相同把主表的ID冲掉
					if (StringUtils.isBlank(parentId)) {// 关联关系没有parent
						errorlog.error(esObjBuilder.toString());// 打错误日志跳过
						continue;
					}
					esObjBuilder.setRelaValue(RelaValue.newBuilder().setName(relaName).setParent(parentId));// tams_relations
				}
			}
			esDataBuilder.addDatas(esObjBuilder.build());
		}

		Result ret = eSClient.docWriteBatch_tc(esDataBuilder.build());
		if (!ret.isSuc()) {
			BulkItemResponse[] retObjs = (BulkItemResponse[]) ret.retObjs();
			for (BulkItemResponse bulkItemResponse : retObjs) {
				if (bulkItemResponse.isFailed()) {
					dataBuilders.getDump().getMetric().counter_send_error.inc();
					log.error(bulkItemResponse.getId());// TODO 错误处理
				}
			}
			LoggerUtil.exit(JvmStatus.s15);
		}
	}
}
